package com.kx.demo.namenode;

import java.util.LinkedList;

/**
 * @author kx
 */
public class FSEditlog {

    /**
     * 双重缓冲
     */
    private DoubleBuffer doubleBuffer;
    /**
     * 唯一id
     */
    private long txid;
    /**
     * 是否正在刷盘
     */
    private volatile Boolean isSyncRunning;
    /**
     * 是否正存在线程在等待刷盘
     */
    private volatile Boolean isWaitingForSync;
    /**
     * 当前刷盘的最大txid
     */
    private volatile long maxTxid;
    /**
     * 当前线程存储的txid
     */
    private ThreadLocal<Long> currentTxid = new ThreadLocal<>();


    public FSEditlog() {
        this.doubleBuffer = new DoubleBuffer();
    }

    /**
     * 多个线程写日志
     *
     * @param content 日志内容
     */
    public void logEdit(String content) {

        // 获取当前对象的锁
        synchronized (this) {
            ++txid;
            Log log = new Log(txid, content);

            currentTxid.set(txid);

            doubleBuffer.write(log);
        }

        logSync();

    }

    /**
     * 将缓冲区信息刷入磁盘
     */
    private void logSync() {

        synchronized (this) {

            // 只需要一个线程去等待就可以
            // 此时存在线程正在刷盘
            if (isSyncRunning) {

                // 当前线程的 txid 正在被另一个线程刷盘
                if (currentTxid.get() <= maxTxid) {
                    return;
                }

                // 是否存在线程在等待刷盘
                if (isWaitingForSync) {
                    return;
                }

                // 当前线程等待去刷盘
                isWaitingForSync = true;
                while (isSyncRunning) {
                    try {
                        // 等待其他线程刷盘完成, 自己先释放当前锁
                        wait(2 * 1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }

                // 执行去刷盘的操作
                isWaitingForSync = false;
            }
            // 交换缓冲区
            doubleBuffer.setReadyToSync();
            // 设置刷盘的最大id
            maxTxid = doubleBuffer.getMaxTxid();
            // 设置正在刷盘
            isSyncRunning = true;
        }

        // 实际操作
        doubleBuffer.flush();

        synchronized (this) {
            isSyncRunning = false;
            notifyAll();
        }
    }


    private class DoubleBuffer {
        /**
         * 写入内存的buffer
         */
        private LinkedList<Log> cacheBuffer = new LinkedList<>();
        /**
         * 写入磁盘的buffer
         */
        private LinkedList<Log> syncBuffer = new LinkedList<>();

        /**
         * 获取正在刷盘的最大 txid
         *
         * @return txid
         */
        public long getMaxTxid() {
            return this.syncBuffer.getLast().getTxid();
        }

        /**
         * 内存写入
         */
        public void write(Log log) {
            this.cacheBuffer.addLast(log);
        }

        /**
         * 交换缓冲区, 做好将 syncBuffer 刷入磁盘的前提
         */
        public void setReadyToSync() {
            LinkedList<Log> tmp = this.cacheBuffer;
            this.cacheBuffer = this.syncBuffer;
            this.syncBuffer = tmp;
        }

        /**
         * 刷入磁盘
         */
        public void flush() {
            for (Log log : syncBuffer) {
                System.out.println("刷入磁盘中: " + log);
            }
            syncBuffer = new LinkedList<>();
        }


        public LinkedList<Log> getCacheBuffer() {
            return cacheBuffer;
        }

        public void setCacheBuffer(LinkedList<Log> cacheBuffer) {
            this.cacheBuffer = cacheBuffer;
        }

        public LinkedList<Log> getSyncBuffer() {
            return syncBuffer;
        }

        public void setSyncBuffer(LinkedList<Log> syncBuffer) {
            this.syncBuffer = syncBuffer;
        }
    }

    /**
     * 日志
     */
    private class Log {
        long txid;
        String content;

        public Log(long txid, String content) {
            this.txid = txid;
            this.content = content;
        }

        public long getTxid() {
            return txid;
        }

        public void setTxid(long txid) {
            this.txid = txid;
        }

        public String getContent() {
            return content;
        }

        public void setContent(String content) {
            this.content = content;
        }

        @Override
        public String toString() {
            return "Log{" +
                    "txid=" + txid +
                    ", content='" + content + '\'' +
                    '}';
        }
    }
}
